package priv.abadstring.study.rocketmq.javaapi.order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * Consumer
 *
 * @author aBadString
 * @date 2021-08-01 16:55:26
 */
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("haoke");
        consumer.setNamesrvAddr("abadstring.me:9876");

        consumer.subscribe("song-song", "*");

        // MessageListenerOrderly 顺序监听. 只会从同一个队列中顺序的拿消息
        consumer.registerMessageListener(new MessageListenerOrderly() {

            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                list.stream()
                        .map(message -> new String(message.getBody(), StandardCharsets.UTF_8))
                        .forEach(System.out::println);
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
    }
}
